package com.zhoucb.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Create 2023-05-01 20:02
 * @Description
 */
public class KfkProducer {
  /*  public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0;i<5;i++){
            kafkaProducer.send(new ProducerRecord<>("first","zhou-"+i));
        }
        kafkaProducer.close();
    }*/
    /*public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 5;i<10;i++){
            kafkaProducer.send(new ProducerRecord<>("first", "zhou-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null){
                        System.out.println("callback: topic:"+recordMetadata.topic()+" partition："+recordMetadata.partition());
                    }
                }
            });
        }
        kafkaProducer.close();
    }*/
  public static void main(String[] args) {
      Properties properties = new Properties();
      properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
      properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
      properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //应答级别
      properties.put(ProducerConfig.ACKS_CONFIG,"1");
      //重试次数
      properties.put(ProducerConfig.RETRIES_CONFIG,3);
      KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
      for (int i = 0;i<5;i++){
          try {
              kafkaProducer.send(new ProducerRecord<>("first","zhou-"+i)).get();
          } catch (InterruptedException e) {
              e.printStackTrace();
          } catch (ExecutionException e) {
              e.printStackTrace();
          }
      }
      kafkaProducer.close();
  }
}
